package cas.ihep.fs.impl;

import alluxio.util.network.NetworkAddressUtils;
import cas.ihep.fs.UnifiedAbstractFileSystem;
import cas.ihep.fs.UnifiedFile;
import cas.ihep.fs.UnifiedFileSystem;
import cas.ihep.fs.alluxio.AlluxioFileSystem;
import cas.ihep.fs.file.LocalFileSystem;
import cas.ihep.fs.hdfs.HdfsFileSystem;
import com.google.common.base.Strings;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hdfs.DFSClient;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.InvocationTargetException;
import java.net.InetAddress;
import java.net.URI;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;

public class UnifiedFileSystemImpl extends UnifiedFileSystem {

    @Override
    public UnifiedFile open(URI path,UnifiedFileSystem.OpenOptions options) throws IOException {
        String scheme=Strings.nullToEmpty(path.getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        return proxy.open(path,options);
    }

    @Override
    public UnifiedFile create(URI path,UnifiedFileSystem.CreateOptions options) throws IOException {
        String scheme=Strings.nullToEmpty(path.getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        return proxy.create(path,options);
    }

    private static final UnifiedFileSystem.CreateOptions DEFAULT_CREATE_OPTION = UnifiedFileSystem.defaultCreateOptions();
    private static final UnifiedFileSystem.OpenOptions DEFAULT_OPEN_OPTION=UnifiedFileSystem.defaultOpenOptions();
    @Override
    public UnifiedFile create(URI path) throws IOException {
        return create(path, DEFAULT_CREATE_OPTION);
    }

    @Override
    public UnifiedFile open(URI uri) throws IOException{
        return open(uri,DEFAULT_OPEN_OPTION);
    }

    public void mkdir(URI uri)throws IOException{
        mkdir(uri,DEFAULT_CREATE_OPTION);
    }

    public void mkdir(URI uri,UnifiedFileSystem.CreateOptions options)throws IOException{
        String scheme=Strings.nullToEmpty(uri.getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        proxy.mkdir(uri,options);
    }

    @Override
    public InputStream openStream(URI path, UnifiedFileSystem.OpenOptions options) throws IOException {
        String scheme=Strings.nullToEmpty(path.getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        return proxy.openStream(path,options);
    }

    @Override
    public OutputStream createStream(URI path, UnifiedFileSystem.CreateOptions options) throws IOException {
        String scheme=Strings.nullToEmpty(path.getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        return proxy.createStream(path,options);
    }

    @Override
    public OutputStream createStream(URI path) throws IOException {
        return createStream(path, DEFAULT_CREATE_OPTION);
    }

    @Override
    public InputStream openStream(URI uri) throws IOException{
        return openStream(uri,DEFAULT_OPEN_OPTION);
    }

    @Override
    public void persist(URI path) throws IOException {
        String scheme=Strings.nullToEmpty(path.getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        proxy.persist(path);
    }

    @Override
    public void persist(UnifiedFile file) throws IOException {
        String scheme=Strings.nullToEmpty(file.path().getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        proxy.persist(file);
    }

    @Override
    public void free(URI path) throws IOException {
        String scheme=Strings.nullToEmpty(path.getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        proxy.free(path);
    }

    @Override
    public void free(UnifiedFile file) throws IOException {
        String scheme=Strings.nullToEmpty(file.path().getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        proxy.free(file);
    }

    @Override
    public void delete(URI path) throws IOException {
        String scheme=Strings.nullToEmpty(path.getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        proxy.delete(path);
    }

    @Override
    public void delete(UnifiedFile file) throws IOException {
        String scheme=Strings.nullToEmpty(file.path().getScheme());
        Proxy proxy=proxies.get(scheme);
        if(proxy==null){
            throw new IOException("Unsupported scheme: "+scheme);
        }
        proxy.delete(file);
    }

    @Override
    public void close() throws IOException {
        if(!isClosed){
            isClosed=true;
            IOException exception=null;
            for(Proxy proxy:proxies.values()){
                try{
                    if(proxy.unifiedFileSystem!=null) {
                        proxy.unifiedFileSystem.close();
                    }
                }catch (IOException e){
                    if(exception==null){
                        exception=new IOException();
                    }
                    exception.addSuppressed(e);
                }
            }
            if(exception!=null){
                throw exception;
            }
        }
    }

    @Override
    public DFSClient dfsClient() {
        try{
            return proxies.get("hdfs").dfsClient();
        }catch (IOException e){
            throw new RuntimeException(e);
        }
    }

    @Override
    public FileSystem hdfsSystem() {
        try{
            return proxies.get("hdfs").hdfsSystem();
        }catch (IOException e){
            throw new RuntimeException(e);
        }
    }

    @Override
    public alluxio.client.file.FileSystem alluxioFileSystem() {
        try {
            return proxies.get("alluxio").alluxioFileSystem();
        }catch (IOException e){
            throw new RuntimeException(e);
        }
    }

    private class Proxy{
        Proxy(Class<? extends UnifiedAbstractFileSystem> cls){
            klass=cls;
            unifiedFileSystem=null;
        }
        Class<? extends UnifiedAbstractFileSystem> klass;
        UnifiedAbstractFileSystem unifiedFileSystem;
        DFSClient dfsClient()throws IOException{
            checkSystem();
            if(unifiedFileSystem instanceof HdfsFileSystem){
                return ((HdfsFileSystem)unifiedFileSystem).getDfsClient();
            }
            throw new RuntimeException("Should never happens");
        }
        FileSystem hdfsSystem() throws IOException {
            checkSystem();
            if(unifiedFileSystem instanceof HdfsFileSystem){
                return ((HdfsFileSystem)unifiedFileSystem).getHdfs();
            }
            throw new RuntimeException("Should never happens");
        }
        alluxio.client.file.FileSystem alluxioFileSystem() throws IOException {
            checkSystem();
            if(unifiedFileSystem instanceof AlluxioFileSystem){
                return ((AlluxioFileSystem)unifiedFileSystem).getFileSystem();
            }
            throw new RuntimeException("Should never happens");
        }
        void checkSystem()throws IOException{
            if(unifiedFileSystem==null){
                synchronized (this){
                    if(unifiedFileSystem==null){
                        try {
                            unifiedFileSystem=klass.getConstructor(UnifiedFileSystem.Configuration.class).newInstance(uConf);
                        } catch (InstantiationException|IllegalAccessException|NoSuchMethodException|InvocationTargetException e) {
                            throw new IOException("Unsupported operation: "+e.getMessage());
                        }
                    }
                }
            }
        }
        UnifiedFile open(URI path, UnifiedFileSystem.OpenOptions options)throws IOException{
            checkSystem();
            return unifiedFileSystem.open(path,options);
        }

        UnifiedFile create(URI path, UnifiedFileSystem.CreateOptions options)throws IOException{
            checkSystem();
            return unifiedFileSystem.create(path,options);
        }

        void free(URI path)throws IOException{
            checkSystem();
            unifiedFileSystem.free(path);
        }

        void free(UnifiedFile file)throws IOException{
            checkSystem();
            unifiedFileSystem.free(file);
        }

        void delete(URI path)throws IOException{
            checkSystem();
            unifiedFileSystem.delete(path);
        }

        void delete(UnifiedFile file)throws IOException{
            checkSystem();
            unifiedFileSystem.delete(file);
        }

        void persist(URI path)throws IOException{
            checkSystem();
            unifiedFileSystem.persist(path);
        }

        void persist(UnifiedFile file)throws IOException{
            checkSystem();
            unifiedFileSystem.persist(file);
        }

        InputStream openStream(URI uri,UnifiedFileSystem.OpenOptions options)throws IOException{
            checkSystem();
            return unifiedFileSystem.openStream(uri,options);
        }

        OutputStream createStream(URI uri,UnifiedFileSystem.CreateOptions options)throws IOException{
            checkSystem();
            return unifiedFileSystem.createStream(uri,options);
        }

        void mkdir(URI uri,UnifiedFileSystem.CreateOptions options)throws IOException{
            checkSystem();
            unifiedFileSystem.mkdir(uri,options);
        }
    }
    private final Map<String,Proxy> proxies;
    private final UnifiedFileSystem.Configuration uConf;
    private boolean isClosed;
    public UnifiedFileSystemImpl(UnifiedFileSystem.Configuration conf){
        proxies=new HashMap<>();
        uConf=conf;
        proxies.put("alluxio",new Proxy(AlluxioFileSystem.class));
        proxies.put("hdfs",new Proxy(HdfsFileSystem.class));
        Proxy local=new Proxy(LocalFileSystem.class);
        proxies.put("",local);
        proxies.put("file",local);
        isClosed=false;
    }

    public static final String localhostName;
    public static final String localhostIP;

    static{
        localhostName= NetworkAddressUtils.getLocalHostName();
        String ip;
        try{
            ip=InetAddress.getLocalHost().getHostAddress();
        }catch(UnknownHostException e){
            e.printStackTrace();
            ip="";
        }
        localhostIP=ip;
    }

}
